#ifndef __MY_CONNECTION__
#define __MY_CONNECTION__

#include "../mqcommon/timer.hpp"
#include "muduo/base/CountDownLatch.h"
#include "muduo/net/TcpClient.h"
#include "muduo/proto/dispatcher.h"

#include "../mqcommon/helpper.hpp"
#include "async_work.hpp"
#include "channel.hpp"

/**
 * 在本项目的客户端设计中，弱化了客户端的概念，而是使用向外提供接口的方式
 * 来提供本项目的服务，所以我们的连接管理实际上就是本项目的客户端
 * 用户只需要使用获得的信道就可以获得所有的功能接口
 * 因为客户端也只需要一个连接来向服务器发起请求，然后收到对应的响应以及订阅的消息即可
 */

namespace NS_Connection
{
    class Connection
    {

        const uint64_t time_interval = 5;

    public:
        using Ptr = std::shared_ptr<Connection>;
        Connection(const std::string &ip, int port, const NS_AsyncWorker::AsyncWorker::Ptr &async_worker)
            : _latch(1), _dispatcher(std::bind(&Connection::defaultMessageCallback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _async_worker(async_worker),
              _client(_async_worker->_loop.startLoop(), muduo::net::InetAddress(ip, port), "Client"),
              _channel_manager(std::make_shared<NS_Channel::ChannelManager>()),
              _timer(std::make_shared<Timer>())
        {
            // DLOG("发送连接");
            _dispatcher.registerMessageCallback<NS_Protocol::BasicCommonResponse>(std::bind(&Connection::basicResponse, this, std::placeholders::_1,
                                                                                            std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<NS_Protocol::BasicConsumerResponse>(std::bind(&Connection::basicConsumerResponse, this, std::placeholders::_1,
                                                                                              std::placeholders::_2, std::placeholders::_3));

            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // DLOG("发送连接");
            _client.connect();
            // DLOG("发送连接成功");
            _latch.wait();
        }

        NS_Channel::Channel::Ptr openChannel()
        {
            // DLOG("开始创建信道");
            auto ptr = _channel_manager->create(_conn, _codec, _async_worker);
            // DLOG("创建信道成功");
            bool ret = ptr->openChannel();
            DLOG("创建信道请求申请成功");
            if (ret == false)
            {
                DLOG("创建信道失败");
                _channel_manager->remove(ptr->cid());
                return NS_Channel::Channel::Ptr();
            }
            // DLOG("创建信道成功");
            return ptr;
        }
        void closeChannel(NS_Channel::Channel::Ptr channel)
        {
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }

        bool connected()
        {
            if (_conn && _conn->connected())
            {
                // DLOG("连接正常");
                return true;
            }
            else
            {
                return false;
            }
        }

        ~Connection() {}

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _conn = _client.connection();
                DLOG("连接建立成功");
                _latch.countDown();
                _timer->push(std::bind(&Connection::sendTimeStamp, this), 0, time_interval, NS_Helper::Uuid::uuid());
            }
            else
            {
                DLOG("连接关闭");
                _timer->stop();
                _conn.reset();
                exit(0);
            }
        }

        void defaultMessageCallback(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            conn->shutdown();
        }

        void basicResponse(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::BasicCommonResponsePtr &message, muduo::Timestamp)
        {
            auto channel_ptr = _channel_manager->get(message->channel_id());
            if (channel_ptr == nullptr)
            {
                DLOG("不存在对应信道");
                return;
            }
            channel_ptr->basicResponse(message);
        }

        void basicConsumerResponse(const muduo::net::TcpConnectionPtr &conn, const NS_Channel::BasicConsumerResponsePtr &message, muduo::Timestamp)
        {
            auto channel_ptr = _channel_manager->get(message->channel_id());
            if (channel_ptr == nullptr)
            {
                DLOG("不存在对应信道");
                return;
            }
            // 2.  封装异步任务（消息处理任务），抛⼊线程池

            // DLOG("channel_ptr引用计数 %ld, message引用计数：%ld", channel_ptr.use_count(), message.use_count());
            // _async_worker->_thraed_pool.post([channel_ptr, message]()
            //                                  { channel_ptr->basicConsumerResponse(message); });

            _async_worker->_thraed_pool.post(std::bind(&NS_Channel::Channel::basicConsumerResponse, channel_ptr.get(), std::move(message)));

            // _async_worker->_thraed_pool.push([channel_ptr, message]()
            //                                  { channel_ptr->basicConsumerResponse(message); });
        }

        void sendTimeStamp()
        {
            if (_conn && _conn->connected())
            {
                NS_Protocol::ConnectionTimestamp req;
                req.set_timestamp(_timer->now());
                DLOG("向服务器发送存活信息，时间戳：%ld", req.timestamp());
                _codec->send(_conn, req);
            }
            // _conn->send()
        }

    private:
        muduo::CountDownLatch _latch;
        NS_AsyncWorker::AsyncWorker::Ptr _async_worker;
        muduo::net::TcpConnectionPtr _conn;
        muduo::net::TcpClient _client;
        ProtobufDispatcher _dispatcher;
        NS_Channel::ProtobufCodecPtr _codec;

        NS_Channel::ChannelManager::Ptr _channel_manager;

        Timer::Ptr _timer;
    };
}

#endif